package kmeans.mapreduce;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import kmeans.bean.WXKmeansCentroids;
import kmeans.bean.WXKmeansData;
import kmeans.exception.WXKmeansException;

import java.io.IOException;

public class WXKmeansMapper extends Mapper<Object, Text, IntWritable, WXKmeansData> {
    private WXKmeansCentroids centroids;
    private int dimension;

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        Configuration configuration = context.getConfiguration();
        dimension = Integer.valueOf(configuration.get("dimension"));
//        System.out.println(dimension);
        centroids = new WXKmeansCentroids(dimension, configuration.get("centroid.path"));
        System.out.println("map 初始化成功");

    }

    @Override
    protected void map(Object key, Text value, Context context) throws IOException, InterruptedException {
        String s = value.toString().trim();
        String[] valuearray = s.split(" ");
//        System.out.println("map操作");
        if (valuearray.length == Integer.valueOf(context.getConfiguration().get("dimension"))) {
            WXKmeansData data = new WXKmeansData(new Text(s), new IntWritable(1));
            try {
                int clusterid = centroids.getClusterid(data);
                context.write(new IntWritable(clusterid), data);
                System.out.println(clusterid +"----"+data.getkMeansText().toString()+"----"+ data.getDataSize());;
            } catch (WXKmeansException e) {
                e.printStackTrace();
            }
        }
    }
}
